package msgbus

import (
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/rs/zerolog/log"
	"strings"
	"time"
)

var topic string
var subTopics []SubTopic

type SubTopic struct {
	topic    string
	callback mqtt.MessageHandler
	qos      byte
}

func Connect(sn string, addr string, username string, password string) bool {
	mqttid = sn
	address = addr

	msgqueue = make(chan interface{}, MSGQUEUE_SIZE)

	url := fmt.Sprintf("tcp://%s", address)

	log.Info().Msgf("start to connect mqtt. %s client id %s", url, mqttid)
	opts := mqtt.NewClientOptions().AddBroker(url).SetClientID(mqttid)

	// 心跳
	opts.SetKeepAlive(60 * time.Second)
	opts.SetPingTimeout(3 * time.Second)

	//// 首次连接时重连，连接时会阻塞当前协程，一直重连
	//opts.SetConnectRetry(true)
	//// 首次连接时重连间隔
	//opts.SetConnectRetryInterval(10 * time.Second)
	// 每次连接的超时时间
	opts.SetConnectTimeout(5 * time.Second)

	// 第一次连接或重连成功后回调
	opts.SetOnConnectHandler(connectHandler)

	// 中断连接后是否重连
	opts.SetAutoReconnect(true)
	// 中断连接后尝试重连时回调
	opts.SetReconnectingHandler(reconnectHandler)
	// 中断连接后的重连间隔
	opts.SetMaxReconnectInterval(5 * time.Second)

	if username != "" {
		opts.Username = username
	}

	if password != "" {
		opts.Password = password
	}

	opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
		msgqueue <- msg
	})

	mqttcli = mqtt.NewClient(opts)
	if token := mqttcli.Connect(); token.Wait() && token.Error() != nil {
		log.Error().Msgf("connect error . %s", token.Error())
		return false
	}

	//订阅本网关的消息
	topic = fmt.Sprintf("/iot/%s/#", mqttid)

	// 默认处理器，兜底的
	Subscribe(topic, 0, nil)

	log.Info().Msgf("mqtt connect mqtt. success. %s", mqttid)
	go loop()
	return true
}

func Subscribe(topic string, qos byte, callback mqtt.MessageHandler) {
	if token := mqttcli.Subscribe(topic, qos, callback); token.Wait() && token.Error() != nil {
		log.Error().Msgf("%s", token.Error())
	} else {
		log.Info().Msgf("mqtt client subscribe %s", topic)
	}

	subTopic := SubTopic{
		topic:    topic,
		callback: callback,
		qos:      qos,
	}
	subTopics = append(subTopics, subTopic)
}

func connectHandler(client mqtt.Client) {
	log.Info().Msgf("mqtt client connected first time or reconnected")

	for _, subTopic := range subTopics {
		mqttcli.Subscribe(subTopic.topic, subTopic.qos, subTopic.callback)
		log.Info().Msgf("mqtt client subscribe %s on connect", subTopic.topic)
	}
}

func reconnectHandler(client mqtt.Client, ops *mqtt.ClientOptions) {
	log.Info().Msgf("mqtt client %s reconnecting", ops.ClientID)
}

func Disconnect() {
	mqttcli.Disconnect(2)
}

func RegisteDispatch(handle DispatchHandle) {
	dispatch = handle
}

type DispatchHandle func(biz string, iot *MsgIoT)

// internal
var (
	MSGQUEUE_SIZE = 1024
)

var (
	address  string
	username string
	password string
	mqttid   string
	mqttcli  mqtt.Client
	msgqueue chan interface{}
	dispatch DispatchHandle
	iotToken *SSIoTToken
	sid      uint64
)

func loop() {
	for {
		p := <-msgqueue
		msg := p.(mqtt.Message)

		log.Trace().Msgf("%s", msg.Topic())

		prefix := fmt.Sprintf("/iot/%s", mqttid)
		if !strings.HasPrefix(msg.Topic(), prefix) {
			continue
		}

		//拆分主题
		topic := msg.Topic()[len(prefix)+1:]
		index := strings.Index(topic, "/")
		//from := topic[:index]
		biz := topic[index+1:]

		iot := new(MsgIoT)
		if err := json.Unmarshal(msg.Payload(), iot); err != nil {
			log.Error().Msgf("%s\n", err)
			continue
		}
		if iot.To == "" {
			iot.To = mqttid
		}

		handleMsg(biz, iot)
	}
}

func handleMsg(biz string, iot *MsgIoT) {
	if iot.Sid > 0 && iotToken != nil && iotToken.req.Sid == iot.Sid {
		//log.Info().Msgf("token %d", iot.Sid)
		iotToken.ack = iot
		iotToken.Done()
		return
	}
	if dispatch != nil {
		dispatch(biz, iot)
	}
}
